package com.kancy.delay.queue.redis.handler;

import com.kancy.delay.queue.redis.message.DelayMessage;
import com.kancy.delay.queue.redis.properties.DelayQueueProperties;
import com.kancy.delay.queue.redis.properties.TopicProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.TaskExecutor;

import java.util.concurrent.TimeUnit;

/**
 * <p>
 * RetryDelayQueueExceptionHandler
 * <p>
 *
 * @author: kancy
 * @date: 2020/3/5 13:27
 **/

public class RetryDelayQueueExceptionHandler implements DelayQueueExceptionHandler{
    private static final Logger log = LoggerFactory.getLogger(RetryDelayQueueExceptionHandler.class);

    private final TaskExecutor taskExecutor;
    private final DelayQueueProperties delayQueueProperties;

    public RetryDelayQueueExceptionHandler(TaskExecutor taskExecutor, DelayQueueProperties delayQueueProperties) {
        this.taskExecutor = taskExecutor;
        this.delayQueueProperties = delayQueueProperties;
    }

    /**
     * 处理异常
     *
     * @param handler
     * @param delayMessage
     * @param exception
     */
    @Override
    public void onProcessMessageException(DelayQueueMessageHandler handler, DelayMessage delayMessage, Exception exception) {
        taskExecutor.execute(() -> retry(handler, delayMessage));
    }

    /**
     * 重试
     * @param handler
     * @param delayMessage
     */
    protected final void retry(DelayQueueMessageHandler handler, DelayMessage delayMessage) {
        TopicProperties topicProperties = delayQueueProperties.getTopics().get(delayMessage.getTopic());
        Integer maxRetry = topicProperties.getMaxRetry();
        boolean retryResult = false;
        int retryTimes = 0;
        while (!retryResult && retryTimes < maxRetry){
            retryTimes ++;
            try {
                delayMessage.setRetry(retryTimes);
                handler.process(delayMessage);
                log.info("Retry [{}] processing topic [{}] delay message [{}] succeeded.",
                        retryTimes, delayMessage.getTopic(), delayMessage.getId());
                retryResult = true;
            } catch (Exception e) {
                log.info("Retry [{}] processing topic [{}] delay message [{}] failed: {}",
                        retryTimes, delayMessage.getTopic(), delayMessage.getId(), e.getMessage());
                try {
                    // 重试间隔
                    TimeUnit.MILLISECONDS.sleep(topicProperties.getRetryInterval());
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
